/*
 * Copyright 2007 Stefan Arentz <http://stefan.arentz.nl>
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.sateh.spaces.simple;

import com.sateh.spaces.Lease;
import com.sateh.spaces.SpaceConnection;
import com.sateh.spaces.SpaceEntry;
import com.sateh.spaces.SpaceException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import javax.transaction.*;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

/**
 * This class implements the proxy for the @{link SimpleSpace} space implementation. It is Stefan's
 * first naive try to implement his own flavour of the JavaSpaces specification.
 *
 * @author Stefan Arentz <stefan@arentz.nl>
 */

@SuppressWarnings({"unchecked"})
public class SimpleSpaceConnection implements SpaceConnection
{
    private final static Log log = LogFactory.getLog(SimpleSpaceConnection.class);

    /**
     *
     */
    
    private final SimpleSpace space;

    /**
     * 
     */

    private final static Map<String,SimpleSpace> spaces = new HashMap<String,SimpleSpace>();

    /**
     *
     */

    private final String name;

    /**
     *
     */

    private TransactionManager transactionManager;
    
    //

    SimpleSpaceConnection(String name)
    {
        this(name, null);
    }

    SimpleSpaceConnection(String name, TransactionManager transactionManager)
    {
        if (name == null) {
            throw new IllegalArgumentException("Argument 'name' cannot be null");
        }

        if (name.length() == 0) {
            throw new IllegalArgumentException("Invalid argument 'spaceUrl'");
        }

        this.name = name;

        this.transactionManager = transactionManager;

        synchronized(spaces) {
            SimpleSpace simpleSpace = spaces.get(name);
            if (simpleSpace == null) {
                simpleSpace = new SimpleSpace();
                spaces.put(name, simpleSpace);
            }
            this.space = simpleSpace;
        }
    }

    //

    public SimpleSpace getSpace()
    {
        return space;
    }

    public String getName()
    {
        return name;
    }

    //
    
    /**
     * @see com.sateh.spaces.Space#write(Object,long)
     */

    public Lease write(Object object, long timeToLive)
        throws SpaceException
    {
        if (object == null) {
            throw new IllegalArgumentException("Argument 'object' cannot be null");
        }

        if (!(object instanceof SpaceEntry)) {
            throw new IllegalArgumentException("Argument 'object' must implement SpaceEntry");
        }

        if (timeToLive < 0) {
            throw new IllegalArgumentException("Argument 'timeToLive' cannot be negative");
        }
        
        try
        {
            SimpleSpaceObject spaceObject = new SimpleSpaceObject(object, timeToLive);

            return new SimpleSpaceLease(
                getSpace(),
                getSpace().writeSpaceObject(spaceObject, timeToLive, getTransactionId()),
                timeToLive
            );
        }

        catch (IOException exception)
        {
            throw new SpaceException("Cannot serialize the object: " + exception.getMessage(), exception);
        }
    }

    //

    /**
     * @see com.sateh.spaces.Space#read(Class, long) 
     */

    public <T extends SpaceEntry> T read(Class<T> objectClass, long timeout)
        throws SpaceException
    {
        checkObjectClassArgument(objectClass);

        Object object = null;

        SimpleSpaceObject spaceObject = getSpace().readOrWaitForSpaceObject(objectClass.getName(), timeout, getTransactionId());
        if (spaceObject != null)
        {
            try {
                object = spaceObject.getMarshalledObject().get();
            } catch (IOException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            } catch (ClassNotFoundException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            }
        }

        return (T) object;
    }

    /**
     * @see com.sateh.spaces.Space#readIfExists(Class, long) 
     */

    public <T extends SpaceEntry> T readIfExists(Class<T> objectClass, long timeout)
        throws SpaceException
    {
        checkObjectClassArgument(objectClass);

        Object object = null;

        SimpleSpaceObject spaceObject = getSpace().readSpaceObject(objectClass.getName(), getTransactionId());
        if (spaceObject != null) {
            try {
                object = spaceObject.getMarshalledObject().get();
            } catch (IOException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            } catch (ClassNotFoundException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            }
        }

        return (T) object;
    }

    /**
     * @see com.sateh.spaces.Space#read(com.sateh.spaces.SpaceEntry, long)  
     */

    public <T extends SpaceEntry> T read(T templateObject, long timeout) throws SpaceException
    {
        Object object = null;

        SimpleSpaceTemplate template = new SimpleSpaceTemplate(templateObject);
        
        SimpleSpaceObject spaceObject = getSpace().readOrWaitForSpaceObject(template, timeout, getTransactionId());
        if (spaceObject != null) {
            try {
                object = spaceObject.getMarshalledObject().get();
            } catch (IOException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            } catch (ClassNotFoundException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            }
        }

        return (T) object;
    }

    /**
     * @see com.sateh.spaces.Space#readIfExists(com.sateh.spaces.SpaceEntry, long)   
     */

    public <T extends SpaceEntry> T readIfExists(T templateObject, long timeout) throws SpaceException
    {
        Object object = null;
        
        SimpleSpaceTemplate template = new SimpleSpaceTemplate(templateObject);

        SimpleSpaceObject spaceObject = getSpace().readSpaceObject(templateObject.getClass().getName(), template, getTransactionId());
        if (spaceObject != null)
        {
            try {
                object = spaceObject.getMarshalledObject().get();
            } catch (IOException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            } catch (ClassNotFoundException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            }
        }

        return (T) object;
    }

    //

    /**
     * @see com.sateh.spaces.Space#take(Class, long)
     */

    public <T extends SpaceEntry> T take(Class<T> objectClass, long timeout) throws SpaceException
    {
        checkObjectClassArgument(objectClass);

        Object object = null;

        SimpleSpaceObject spaceObject = getSpace().takeOrWaitForSpaceObject(objectClass.getName(), timeout, getTransactionId());
        if (spaceObject != null)
        {
            // What we really need to do here is to roll back the take if the deserialization fails. Maybe all this
            // does need to be moved into the space after all. Or, we remove the object for real when the
            // deserialization was succesful. For now we simply assume success.

            try {
                object = spaceObject.getMarshalledObject().get();
            } catch (IOException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            } catch (ClassNotFoundException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            }
        }

        return (T) object;
    }

    /**
     * @see com.sateh.spaces.Space#takeIfExists(Class, long)
     */

    public <T extends SpaceEntry> T takeIfExists(Class<T> objectClass, long timeout)
        throws SpaceException
    {
        checkObjectClassArgument(objectClass);

        Object object = null;

        SimpleSpaceObject spaceObject = getSpace().takeSpaceObject(objectClass.getName(), getTransactionId());
        if (spaceObject != null)
        {
            // What we really need to do here is to roll back the take if the deserialization fails. Maybe all this
            // does need to be moved into the space after all. Or, we remove the object for real when the
            // deserialization was succesful. For now we simply assume success.

            try {
                object = spaceObject.getMarshalledObject().get();
            } catch (IOException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            } catch (ClassNotFoundException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            }
        }

        return (T) object;
    }

    /**
     * @see com.sateh.spaces.Space#take(com.sateh.spaces.SpaceEntry, long)
     */

    public <T extends SpaceEntry> T take(T templateObject, long timeout)
        throws SpaceException
    {
        Object object = null;

        SimpleSpaceTemplate template = new SimpleSpaceTemplate(templateObject);

        SimpleSpaceObject spaceObject = getSpace().takeOrWaitForSpaceObject(templateObject.getClass().getName(), template, timeout, getTransactionId());
        if (spaceObject != null)
        {
            // What we really need to do here is to roll back the take if the deserialization fails. Maybe all this
            // does need to be moved into the space after all. Or, we remove the object for real when the
            // deserialization was succesful. For now we simply assume success.

            try {
                object = spaceObject.getMarshalledObject().get();
            } catch (IOException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            } catch (ClassNotFoundException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            }
        }

        return (T) object;
    }

    /**
     * @see com.sateh.spaces.Space#takeIfExists(com.sateh.spaces.SpaceEntry, long)
     */

    public <T extends SpaceEntry> T takeIfExists(T templateObject, long timeout)
        throws SpaceException
    {
        Object object = null;

        SimpleSpaceTemplate template = new SimpleSpaceTemplate(templateObject);

        SimpleSpaceObject spaceObject = getSpace().takeSpaceObject(templateObject.getClass().getName(), template, getTransactionId());
        if (spaceObject != null)
        {
            // What we really need to do here is to roll back the take if the deserialization fails. Maybe all this
            // does need to be moved into the space after all. Or, we remove the object for real when the
            // deserialization was succesful. For now we simply assume success.

            try {
                object = spaceObject.getMarshalledObject().get();
            } catch (IOException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            } catch (ClassNotFoundException exception) {
                throw new SpaceException("Cannot read object: " + exception.getMessage(), exception);
            }
        }

        return (T) object;
    }

    // Count

    /**
     * @see com.sateh.spaces.Space#count(Class)
     */

    public long count(Class<? extends SpaceEntry> objectClass) throws SpaceException
    {
        checkObjectClassArgument(objectClass);
        return getSpace().count(objectClass.getName(), getTransactionId());
    }

    /**
     * @see com.sateh.spaces.Space#count(com.sateh.spaces.SpaceEntry) 
     */

    public long count(SpaceEntry templateObject) throws SpaceException
    {
        checkTemplateObjectArgument(templateObject);
        return getSpace().count(new SimpleSpaceTemplate(templateObject), getTransactionId());
    }
    
    //

    private void checkObjectClassArgument(Class<? extends SpaceEntry> objectClass)
    {
        if (objectClass == null) {
            throw new IllegalArgumentException("Argument 'objectClass' cannot be null");
        }
    }

    private void checkTemplateObjectArgument(SpaceEntry templateObject)
    {
        if (templateObject == null) {
            throw new IllegalArgumentException("Argument 'templateObject' cannot be null");
        }
    }    

    //

    private Transaction getTransaction() throws SpaceException
    {
        try {
            if (transactionManager != null) {
                return transactionManager.getTransaction();
            } else {
                return null;
            }
        } catch (SystemException e) {
            throw new SpaceException(e);
        }
    }
    
    private Map<Transaction, TransactionalContext> transactionalContexts = new HashMap<Transaction, TransactionalContext>();

    private TransactionalContext getOrCreateTransactionalContext(Transaction tx) throws SystemException, RollbackException
    {
        TransactionalContext context = null;

        if (transactionManager != null) {
            context = transactionalContexts.get(tx);
            if (context == null) {
                context = new TransactionalContext();
                transactionalContexts.put(tx, context);
                transactionManager.getTransaction().registerSynchronization(context);
            }
        }

        return context;
    }

    private SimpleSpaceTransactionId getTransactionId() throws SpaceException
    {
        Transaction tx = getTransaction();

        if (tx != null) {
            try {
                TransactionalContext transactionalContext = getOrCreateTransactionalContext(tx);
                return transactionalContext.getTransactionId();
            } catch (SystemException e) {
                throw new SpaceException();
            } catch (RollbackException e) {
                throw new SpaceException();
            }
        } else {
            return null;
        }
    }

    class TransactionalContext implements Synchronization
    {
        private final SimpleSpaceTransactionId transactionId = new SimpleSpaceTransactionId();

        public TransactionalContext()
        {
        }

        public SimpleSpaceTransactionId getTransactionId()
        {
            return transactionId;
        }

        /**
         * @see javax.transaction.Synchronization#afterCompletion(int) 
         */

        public void afterCompletion(int status)
        {
            log.debug("afterCompletion; transactionId=" + transactionId + " status=" + status);

            if (status == Status.STATUS_COMMITTED) {
                log.debug("Commit of transaction; " + transactionId);
                space.commitTransaction(transactionId);
            } else if (status == Status.STATUS_ROLLEDBACK) {
                log.debug("Rollback of transaction; " + transactionId);
                space.rollbackTransaction(transactionId);                
            }
        }

        /**
         * @see javax.transaction.Synchronization#beforeCompletion() 
         */

        public void beforeCompletion()
        {
            log.debug("beforeCompletion; transactionId=" + transactionId);
        }
    }
}
